package com.djl.rabbitmq.four;

import com.djl.rabbitmq.utils.RabbitMqUtils;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmCallback;

import java.nio.charset.StandardCharsets;
import java.util.UUID;
import java.util.concurrent.ConcurrentLinkedDeque;
import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ConcurrentSkipListMap;

/**
 * @program: rabbitmq-hello
 * @description: 发布确认模式
 * @author: Mr.dai
 * @create: 2021-09-10 15:40
 **/

public class ConfirmMessage {

  public static void main(String[] args) throws Exception {
  //  单个确认 42750ms
  //  publishMessage();
  //批量确认 50605ms
  //  publishMessageEvery();
  //  异步确认
    publishMessageAsync();
  }

  //批量发消息的个数
    public static final int MESSAGE_COUNT=1000;
  //单个确认
  public static void publishMessage() throws Exception {
    Channel channel = RabbitMqUtils.getChannel();
    String queueName = UUID.randomUUID().toString();
    channel.queueDeclare(queueName,true,false,false,null );
  //  开启发布确认
    channel.confirmSelect();
  //  记录开始的时间
    long begin=System.currentTimeMillis();
    for (int i = 0; i < MESSAGE_COUNT; i++) {
      String msg=i+"";
      channel.basicPublish("",queueName,null,msg.getBytes(StandardCharsets.UTF_8));
      //单个消息确认
      boolean falg = channel.waitForConfirms();
      if (falg==true){
        System.out.println("消息发送成功");
      }
    }
    long end =System.currentTimeMillis();
    System.out.println("发布"+MESSAGE_COUNT+"耗时："+(end-begin)+"ms");




  }
//  批量发送确认
  public static void publishMessageEvery() throws Exception {
    Channel channel = RabbitMqUtils.getChannel();
    String queueName = UUID.randomUUID().toString();
    channel.queueDeclare(queueName,true,false,false,null );
    //  开启发布确认
    channel.confirmSelect();
    //  记录开始的时间
    long begin=System.currentTimeMillis();
    //批量确认大小
    int batchSize=100;

    for (int i = 0; i < MESSAGE_COUNT; i++) {
      String msg=i+"";
      channel.basicPublish("",queueName,null,msg.getBytes(StandardCharsets.UTF_8));
      boolean falg = channel.waitForConfirms();
      //  判断达到一百条消息的时候确认一次
      if(i%(batchSize+1)==0){
        if (falg==true){
          System.out.println("消息发送成功");

        }
      }
    }
    //单个消息确认

    long end =System.currentTimeMillis();
    System.out.println("发布"+MESSAGE_COUNT+"耗时："+(end-begin)+"ms");

  }
  //异步发布确认
  public static void publishMessageAsync() throws Exception {
    Channel channel = RabbitMqUtils.getChannel();
    String queueName = UUID.randomUUID().toString();
    channel.queueDeclare(queueName,true,false,false,null );
    //  开启发布确认
    channel.confirmSelect();
    /**
     * 线程安全有序的一个哈希表 适用于高并发的情况下
     *   1.将序号与消息进行关联
     *   2.批量删除条目，只要给到序号
     *   3.支持高并发
     */
    ConcurrentSkipListMap<Long,String> outstandingConfirms=new ConcurrentSkipListMap<>();


    //  记录开始的时间
    long begin=System.currentTimeMillis();
    //批量确认大小
    int batchSize=100;


    //消息确认成功回调函数
    ConfirmCallback ackCallback= ( deliveryTag, multiple)->{
      //删除已确认的消息，剩下的就是未确认的消息
      System.out.println("确认的消息："+deliveryTag);
      if (multiple){
        ConcurrentNavigableMap<Long, String> confirmed = outstandingConfirms.headMap(deliveryTag);
          confirmed.clear();
      }else {
        outstandingConfirms.remove(deliveryTag);
      }
    };


    //消息确认失败回到函数
    /**
     * 1.消息的参数
     * 2.是否为批量确认
     */
    ConfirmCallback nackCallback =( deliveryTag,  ultiple)->{
      //打印未确认的消息
      String msg = outstandingConfirms.get(deliveryTag);
      System.out.println("未确认的消息："+deliveryTag+"消息名称:"+msg);
    };

    //准备消息的监听器，监听哪些消息失败了 使用两参构造方法
    channel.addConfirmListener(ackCallback,nackCallback);//异步通知
    for (int i = 0; i < MESSAGE_COUNT; i++) {
      String msg=i+"";
      outstandingConfirms.put(channel.getNextPublishSeqNo(),msg);
      channel.basicPublish("",queueName,null,msg.getBytes(StandardCharsets.UTF_8));
    //  记录所有发送的消息
    }

    //单个消息确认

    long end =System.currentTimeMillis();
    System.out.println("异步发布"+MESSAGE_COUNT+"耗时："+(end-begin)+"ms");

  }
}
